package io.gitee.icefairy.swclient4flink;

import org.apache.flink.core.fs.*;
import seaweedfs.client.FilerClient;
import seaweedfs.client.FilerProto;

import java.io.IOException;
import java.net.URI;
import java.util.function.Function;

public class SwFileSystem extends FileSystem {
    private FilerClient fc;
    String workingDir = "/";

    public SwFileSystem(FilerClient fc) {
        this.fc = fc;
    }

    @Override
    public Path getWorkingDirectory() {
        return new Path(workingDir, "");
    }

    @Override
    public Path getHomeDirectory() {
        return new Path(workingDir, "");
    }

    @Override
    public URI getUri() {
        return URI.create("sw://");
    }

    @Override
    public FileStatus getFileStatus(Path f) throws IOException {
        return new SwFileStatus(fc.lookupEntry(f.getParent().getPath(), f.getName()));
    }

    @Override
    public BlockLocation[] getFileBlockLocations(FileStatus file, long start, long len) throws IOException {
        return new BlockLocation[0];
    }

    @Override
    public FSDataInputStream open(Path f, int bufferSize) throws IOException {
        return new SwInputStream(fc, f.getPath());
    }

    @Override
    public FSDataInputStream open(Path f) throws IOException {
        return new SwInputStream(fc, f.getPath());
    }

    @Override
    public FileStatus[] listStatus(Path f) throws IOException {
        return (FileStatus[]) fc.listEntries(f.getPath()).stream().map(new Function<FilerProto.Entry, Object>() {
            @Override
            public Object apply(FilerProto.Entry entry) {
                return new SwFileStatus(entry);
            }
        }).toArray();
    }

    @Override
    public boolean delete(Path f, boolean recursive) throws IOException {
        return fc.rm(f.getPath(), recursive, true);
    }

    @Override
    public boolean mkdirs(Path f) throws IOException {
        return fc.mkdirs(f.getPath(), 666);
    }

    @Override
    public FSDataOutputStream create(Path f, WriteMode overwriteMode) throws IOException {
        return new SwOutputStream(fc, f.getPath());
    }

    @Override
    public boolean rename(Path src, Path dst) throws IOException {
        return fc.mv(src.getPath(), dst.getPath());
    }

    @Override
    public boolean isDistributedFS() {
        return true;
    }

    @Override
    public FileSystemKind getKind() {
        return FileSystemKind.FILE_SYSTEM;
    }
}
